package it.uniroma3.dia.db;


import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/** 
*
* @author Daniele D'Andrea
* @author Antonio Gallo
*/
public class DBItemSimilarity {

	public static ConcurrentLinkedQueue<Integer> concurredLinkedQueueCommon = new ConcurrentLinkedQueue<Integer>();
	public static ConcurrentLinkedQueue<Integer> concurredLinkedQueueSimilarity = new ConcurrentLinkedQueue<Integer>();
	public static CountDownLatch countDownLatch;
	
	/**
	 * @param args
	 * @throws SQLException 
	 */
	public static void main(String[] args) throws Exception {
		
		int movieIDmin = 0;
		
		String MOVIEIDS = "SELECT id FROM movies where id > ? order by id";
		
		Connection connection = new DBDatasource().getConnection();

		PreparedStatement statement = connection.prepareStatement(MOVIEIDS);
		statement.setInt(1, movieIDmin);
		ResultSet result = statement.executeQuery();
		
		String USER_STATS = "insert into user_stats_ratedmovies (userID, mean) " +
				"select userID, sum(rating) / count(*) " +
				"from user_ratedmovies " +
				"group by userID";
		
		
		PreparedStatement statement_similarity;
		statement_similarity = connection.prepareStatement(USER_STATS);
		statement_similarity.execute();
		
		while(result.next()){
			int movieID = result.getInt(1);
			concurredLinkedQueueCommon.offer(Integer.valueOf(movieID));
			concurredLinkedQueueSimilarity.offer(Integer.valueOf(movieID));
		}

		int numberOfThread = Runtime.getRuntime().availableProcessors()*2;
		countDownLatch = new CountDownLatch(numberOfThread);
		ExecutorService executor = Executors.newFixedThreadPool(numberOfThread);
		

		CommonFiller commonFiller = new CommonFiller();
		executor.submit(commonFiller);

		SimilarityFiller similarityFiller = new SimilarityFiller();
		executor.submit(similarityFiller);
		
		countDownLatch.await();
		executor.shutdown();
		
		String INSERT_ALL_INTO_STATS = "insert into movie_item_similarity_stats " +
				"(movieID, neighborID,pearsonCorrelation,numberOfCommonRating,numberOfCommonGenres,numberOfCommonActors,numberOfCommonTags,numberOfCommonDirectors)" +
				"select mis.*, " +
				"IFNULL(g.numberOfCommonGenres,0) as numberOfCommonGenres, " +
				"IFNULL(a.numberOfCommonActors,0) as numberOfCommonActors, " +
				"IFNULL(t.numberOfCommonTags,0) as numberOfCommonTags, " +
				"IFNULL(d1.directorID=d2.directorID,0) as numberOfCommonDirectors from " +
				"movie_item_similarity mis " +
				"left outer join movie_genre_common g " +
				"on g.movieID=mis.movieID and g.neighborID=mis.neighborID " +
				"left outer join movie_actor_common a " +
				"on a.movieID=mis.movieID and a.neighborID=mis.neighborID " +
				"left outer join movie_tags_common t " +
				"on t.movieID=mis.movieID and t.neighborID=mis.neighborID " +
				"left outer join movie_directors d1 " +
				"on mis.movieID=d1.movieID " +
				"left outer join movie_directors d2 " +
				"on mis.neighborID=d2.movieID";

		System.out.println("creating movie_item_similarity_stats...");
		
		statement_similarity = connection.prepareStatement(INSERT_ALL_INTO_STATS);
		statement_similarity.execute();
		
		System.out.println("movie_item_similarity_stats created successfully");
		
		result.close();
		connection.close();
	}
}

class SimilarityFiller implements Runnable{
	
	Connection connection;
	
	public SimilarityFiller() throws Exception{
		connection = new DBDatasource().getConnection();
	}
	
	@Override
	public void run() {
		int count = DBItemSimilarity.concurredLinkedQueueSimilarity.size();
		
		while(count>0){
			Integer _movieID = DBItemSimilarity.concurredLinkedQueueSimilarity.poll();
			if(_movieID!=null){
				int movieID = _movieID.intValue();
				fillCommons(movieID);
			}
			count = DBItemSimilarity.concurredLinkedQueueSimilarity.size();
		}
		
		//TODO: cancella quelli con correlazione di pearson<=0 e null, non sono utili.
		//delete FROM MovieDB.movie_item_similarity
		//where pearsonCorrelation<=0 or pearsonCorrelation is NULL;
		
		DBItemSimilarity.countDownLatch.countDown();
		System.out.println("Thread SimilarityFiller has finished");
		
		try {
			connection.close();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}
	
	private void fillCommons(int movieID) {
		
		String SIMILARITY = "insert into movie_item_similarity" +
				"(movieID, neighborID, pearsonCorrelation, numberOfCommonRating) " +
				"select sf.movieID as itemID, sf.neighborID," +
				"sf.pearsonNum/sf.pearsonDen AS pearsonCorrelation," +
				"sf.numberOfRating " +
				"from (" +
				"select r1.movieID AS movieID, r2.movieID AS neighborID, count(*) as numberOfRating," +
				"SUM( (r1.rating-stats1.mean) * (r2.rating-stats1.mean) ) as pearsonNum," +
				"( SQRT( SUM(POW((r1.rating-stats1.mean),2)) ) ) * SQRT( SUM(POW((r2.rating-stats1.mean),2)) ) as pearsonDen " +
				"from user_ratedmovies r1 " +
				"join user_ratedmovies r2 " +
				"on r1.userID = r2.userID " +
				"join user_stats_ratedmovies stats1 " +
				"on stats1.userID = r1.userID " +
				"where r1.movieID<>r2.movieID " +
				"and r1.movieID = ? " +
				"group by r1.movieID, r2.movieID) sf";
		
		try{
			PreparedStatement statement_similarity = connection.prepareStatement(SIMILARITY);
			statement_similarity.setInt(1, movieID);
			statement_similarity.execute();
		System.out.println("similarity created for movieID: "+movieID);
		}catch(SQLException e){
			e.printStackTrace();
		}
	}
}


class CommonFiller implements Runnable{
	
	Connection connection;
	
	public CommonFiller() throws Exception{
		connection = new DBDatasource().getConnection();
	}
	
	@Override
	public void run() {
		int count = DBItemSimilarity.concurredLinkedQueueCommon.size();
		
		while(count>0){
			Integer _movieID = DBItemSimilarity.concurredLinkedQueueCommon.poll();
			if(_movieID!=null){
				int movieID = _movieID.intValue();
				fillCommons(movieID);
			}
			count = DBItemSimilarity.concurredLinkedQueueCommon.size();
		}
		
		DBItemSimilarity.countDownLatch.countDown();
		System.out.println("Thread CommonFiller has finished");
		
		try {
			connection.close();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}
	
	private void fillCommons(int movieID) {
		
		String COMMON_GENRE = "insert into movie_genre_common SELECT a.id, b.id, IFNULL(count(a.genre),0) as numberOfCommonGenres FROM (select m1.id, g.genre from movies m1 inner join movie_genres g on m1.id=g.movieID where m1.id = ?) a LEFT OUTER JOIN (select m1.id, g.genre from movies m1 inner join movie_genres g on m1.id=g.movieID ) b USING (genre) where a.id<>b.id group by a.id, b.id";
		String COMMON_ACTORS = "insert into movie_actor_common SELECT a.id, b.id, IFNULL(count(a.actorid),0) as numberOfCommonActors FROM (select m1.id, g.actorid from movies m1 inner join movie_actors g on m1.id=g.movieID where m1.id = ?) a LEFT OUTER JOIN (select m1.id, g.actorid from movies m1 inner join movie_actors g on m1.id=g.movieID ) b USING (actorid) where a.id<>b.id group by a.id, b.id";
		String COMMON_TAGS = "insert into movie_tags_common SELECT a.id, b.id, count(a.tagid) as numberOfCommonTags FROM (select m1.id, g.tagid from movies m1 inner join movie_tags g on m1.id=g.movieID where m1.id = ?) a LEFT OUTER JOIN (select m1.id, g.tagid from movies m1 inner join movie_tags g on m1.id=g.movieID ) b USING (tagid) where a.id<>b.id group by a.id, b.id";
		
		String[] queries = new String[]{COMMON_GENRE, COMMON_ACTORS, COMMON_TAGS};
		
		for(String query : queries){
			PreparedStatement statement;
			try {
				statement = connection.prepareStatement(query);
				statement.setInt(1, movieID);
				statement.execute();
				
				statement.close();
			} catch (SQLException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		System.out.println("common updated for movieID: "+movieID);
	}
}
